
package org.zjt.rabbit.demo.rpc;
import org.zjt.rabbit.demo.common.ContantsConfig;
import com.rabbitmq.client.*;

import java.io.IOException;

public class RPCServer {

	private static final String RPC_QUEUE_NAME = "rpc_queue";

	public static void main(String[] argv)throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(ContantsConfig.MQ_HOST);
		Connection connection = null;
		try {
			connection = factory.newConnection();
			final Channel channel = connection.createChannel();
			channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
			channel.basicQos(1);
			System.out.println(" [x] Awaiting RPC requests");

			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
					AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();
					String response = "";
					try {
						String message = new String(body,"UTF-8");
						int n = Integer.parseInt(message);
						System.out.println(" [.] fib(" + message + ")");
						response += fib(n);
					} catch (RuntimeException e){
						System.out.println(" [.] " + e.toString());
					} finally {
						channel.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
						channel.basicAck(envelope.getDeliveryTag(), false);
					}
				}
			};

			channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

		}catch (Exception _e){
			_e.printStackTrace();
		}
	}
	private static int fib(int n) {
		if (n == 0) return 0;
		if (n == 1) return 1;
		return fib(n-1) + fib(n-2);
	}
}
